package week03;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HelloHBase {

    /**
     * 创建表
     * @param conn
     * @param tableName
     * @param columnFamilyAry
     * @throws IOException
     */
    public static void createTable(Connection conn, String tableName, String[] columnFamilyAry) throws IOException {
        Admin admin = conn.getAdmin();
        TableName tname = TableName.valueOf(tableName);
        if (admin.tableExists(tname )){
        }else {
            HTableDescriptor tabDesc =  new HTableDescriptor(tname);
            for (String s : columnFamilyAry) {
                HColumnDescriptor family = new HColumnDescriptor(s);
                tabDesc.addFamily(family);
            }
            admin.createTable(tabDesc);
        }
        admin.close();
    }

    /**
     * 放入数据
     * @param conn
     * @param tableName
     * @param rowKey
     * @param cellAry
     * @throws IOException
     */
    public static void putData(Connection conn, String tableName, String rowKey ,  List<CellValue> cellAry) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(rowKey));
        for (CellValue cv : cellAry) {
            put.addColumn(Bytes.toBytes(cv.getFamily()), Bytes.toBytes(cv.getColumn()), Bytes.toBytes(cv.getValue()));
        }
        table.put(put);
        table.close();
    }

    /**
     * 查询数据
     * @param conn
     * @param tableName
     * @param rowKey
     * @param cell
     * @return
     * @throws IOException
     */
    public static String getData(Connection conn, String tableName, String rowKey, CellValue cell) throws IOException {
        String cellData = null;
        Table table = conn.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(cell.getFamily()), Bytes.toBytes(cell.getColumn()));
        Result rst = table.get(get);
        cellData = rst.toString();
        table.close();
        return cellData;
    }

    /**
     * 删除数据
     * @param conn
     * @param tableName
     * @param rowKey
     * @param cell
     * @throws IOException
     */
    public static void deleteData(Connection conn, String tableName, String rowKey,CellValue cell) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Delete del = new Delete(Bytes.toBytes(rowKey));
        del.addColumn(Bytes.toBytes(cell.getFamily()),Bytes.toBytes(cell.getColumn()));
        table.delete(del);
        table.close();
    }

    /**
     * 删除表
     * @param conn
     * @param tableName
     * @throws IOException
     */
    public static void deleteTable(Connection conn, String tableName) throws IOException {
        Admin admin = conn.getAdmin();
        TableName tname = TableName.valueOf(tableName);
        admin.disableTable(tname);
        admin.deleteTable(tname);
        admin.close();
    }

    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","127.0.0.1");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        conf.set("hbase.client.username","root");
        conf.set("hbase.client.password","root");
        Connection conn = ConnectionFactory.createConnection(conf);

        String tableName = "Sunzihao_student";
        String family_name = "name";
        String family_info = "info";
        String family_score = "score";
        String name_nm = "";
        String info_stuId = "student_id";
        String info_class = "class";
        String score_und = "understanding";
        String score_prog = "programming";
        String[] familyAry = {family_name,family_info,family_score};
        createTable(conn,tableName,familyAry);

        String row1 = "Tom,20220000000001,1,75,82";
        String row2 = "Jerry,20220000000002,1,85,67";
        String row3 = "Jack,20220000000003,2,80,80";
        String row4 = "Rose,20220000000004,2,60,61";
        String row5 = "Sunzihao,G20220735020121,2,75,69";
        String[] rowAry = {row1,row2,row3,row4,row5};

        for (String line : rowAry) {
            String[] tokenAry = line.split(",", -1);
            List<CellValue> cellList = new ArrayList<>();

            CellValue nmCell = new CellValue(family_name,name_nm,tokenAry[0]);
            cellList.add(nmCell);

            CellValue stuIdCell = new CellValue(family_info,info_stuId,tokenAry[1]);
            cellList.add(stuIdCell);

            CellValue classCell = new CellValue(family_info,info_class,tokenAry[2]);
            cellList.add(classCell);

            CellValue undCell = new CellValue(family_score,score_und,tokenAry[3]);
            cellList.add(undCell);

            CellValue progCell = new CellValue(family_score,score_prog,tokenAry[4]);
            cellList.add(progCell);

            String rowKey = nmCell.getValue() + "-" + stuIdCell.getValue();
            putData(conn,tableName,rowKey,cellList);
        }

        conn.close();
    }
}
